package cn.fanqinghong.ymcc.mq.consumer;

import cn.fanqinghong.ymcc.constants.CourseOrderContants;
import cn.fanqinghong.ymcc.domain.PayOrder;
import cn.fanqinghong.ymcc.dto.PayDto;
import cn.fanqinghong.ymcc.service.IPayOrderService;
import cn.fanqinghong.ymcc.util.AssertUtil;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.JsonObject;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.Date;

/**
 * FAN
 * 一个消息只能生成一个支付单
 * 2022/8/1  9:36
 */
@RocketMQMessageListener(
        consumerGroup = CourseOrderContants.COURSE_ORDER_PRODUCER_GROUP,
        topic = CourseOrderContants.COURSE_ORDER_PRODUCER_TOPIC,
        selectorExpression = CourseOrderContants.COURSE_ORDER_PRODUCER_TAG
        //messageModel = MessageModel.CLUSTERING//默认集群模式，一个消息只会被一个消费者消费
)
@Component
public class CourseOrder2PayCourseConsumer implements RocketMQListener<MessageExt> {
    @Autowired
    private IPayOrderService payOrderService;
    /**
     * 必须保证消费方的幂等性
     * 1.拿到消息，检查消息不为空
     * 2.根据拿到的消息，查询PyaOrder表，校验订单号对应的数据
     * 2.1 有，证明已经消费过了，直接签收消息  mq就不会推送这个消息了
     * 2.2 没有，表示第一次消费，保存payOrder表
     * @param messageExt
     */
    @Override
    public void onMessage(MessageExt messageExt) {
        byte[] body = messageExt.getBody();
        AssertUtil.isNotNull(body,"消息发送失败");
        String msgStr = new String(body, StandardCharsets.UTF_8);
        //2.根据拿到的消息，查询PyaOrder表，校验订单号对应的数据
        PayDto payDto = JSONObject.parseObject(msgStr, PayDto.class);
        synchronized (payDto.getOrderNo()) {//加锁，只允许一个线程有一个数据进入
            PayOrder payOrder = payOrderService.selectByOrderNo(payDto.getOrderNo());
            if (payOrder != null) {
                //2.1 有，证明已经消费过了，直接签收消息  mq就不会推送这个消息了
                return;//直接签收了
            }
            //2.2 没有，表示第一次消费，保存payOrder表
            payOrder = new PayOrder();
            BeanUtils.copyProperties(payDto, payOrder);
            payOrder.setCreateTime(new Date());
            payOrder.setPayStatus(PayOrder.COURSE_ORDER_WAITING_TO_PAY);
            payOrderService.insert(payOrder);
        }
    }
}
